package com.rtf.framework.mq.kafka.consumer;

import java.util.List;

import org.apache.commons.collections.CollectionUtils;
import org.springframework.util.Assert;

import com.rtf.framework.mq.common.CommonConstants;
import com.rtf.framework.mq.common.ConsumerProperties;
import com.rtf.framework.mq.common.Service;
import com.google.common.collect.Lists;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class KafkaConsumerConfiger implements Service {

	private KafkaConsumerProperties properties;

	private List<ConsumerProperties> consumerProperties;

	/** The consumers. */
	private List<KafkaConsumerAdaptor> consumers = Lists.newArrayList();

	public KafkaConsumerConfiger(KafkaConsumerProperties properties, List<ConsumerProperties> cosumer) {
		Assert.notNull(properties, "KafkaConsumerProperties is null");
		Assert.notEmpty(cosumer, "Kafka consumer is empty");
		this.consumerProperties = cosumer;
		this.properties = properties;
	}

	@Override
	public void init() {
		// TODO Auto-generated method stub

	}

	@Override
	public void start() {
		log.info("##############begin create kafka consumer....#############");

		KafkaConsumerAdaptor adaptor;
		for (ConsumerProperties consumer : this.consumerProperties) {
			if (CommonConstants.MQ_TYPE_KAFKA.equals(consumer.getType())) {
				for (int i = 0; i < consumer.getCount(); i++) {
					adaptor = new KafkaConsumerAdaptor(this.properties, consumer);
					adaptor.init();
					adaptor.start();
					consumers.add(adaptor);
				}

			}
		}
		log.info("############## create kafka {} consumer end #############", consumers.size());

	}

	@Override
	public void destroy() {
		if (CollectionUtils.isNotEmpty(this.consumers)) {
			for (KafkaConsumerAdaptor adaptor : this.consumers) {
				adaptor.destroy();
			}
		}

	}

}
